package lib.quasar.base.rx;

import java.util.HashMap;

import io.reactivex.Observable;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;

/**
 * description: 线程间通讯, 基于Rxjava2
 * created by kalu on 2017/5/15 15:59
 */
public class RxBus {

    private final Subject<Object> mSubject;
    private HashMap<String, CompositeDisposable> mSubjectHashMap;

    private RxBus() {
        //StickyEvent功能的添加
        mSubject = PublishSubject.create().toSerialized();
    }

    /**
     * ************************************ 单例 **********************************************
     */

    public static RxBus getInstance() {
        return RxBus.SingletonHolder.instance;
    }

    /**
     * 发送一个新事件
     *
     * @param o
     */
    public void post(Object o) {
        mSubject.onNext(o);
    }

    /**
     * 返回特定类型的被观察者
     *
     * @param clazz
     * @param <T>
     * @return
     */
    public <T> Observable<T> toObservable(Class<T> clazz) {
        return mSubject.ofType(clazz);
    }

    /**
     * 是否已经有观察值订阅
     *
     * @return
     */
    public boolean hasObservers() {
        return mSubject.hasObservers();
    }

    /**
     * 一个默认的订阅方法
     */
    public <T> Disposable doSubscribe(Class<T> type, Consumer<? super T> onNext, Consumer<? super Throwable> onError) {
        Disposable subscribe = toObservable(type)
                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(onNext, onError);

        addSubscription(type, subscribe);
        return subscribe;
    }

    /**
     * 一个默认的订阅方法
     */
    public <T> Disposable doSubscribe(Class<T> type, Consumer<? super T> onNext) {
        Disposable subscription = toObservable(type)
                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(onNext);

        addSubscription(type, subscription);
        return subscription;
    }

    /**
     * 一个默认的订阅方法
     */
    public <T> Disposable doSubscribe(Observable<T> observable, Class<T> type, Consumer<? super T> onNext) {
        Disposable subscription = observable
                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(onNext);

        addSubscription(type, subscription);
        return subscription;
    }

    /**
     * 一个默认的订阅方法
     */
    public <T> Disposable doSubscribe(Observable<T> observable, Class<T> type, Consumer<? super T> onNext, Consumer<? super Throwable> onError) {
        Disposable subscription = observable
                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(onNext, onError);

        addSubscription(type, subscription);
        return subscription;
    }

    /**
     * 保存订阅后的subscription
     */
    private <T> void addSubscription(Class<T> type, Disposable subscription) {
        if (mSubjectHashMap == null) {
            mSubjectHashMap = new HashMap<>();
        }
        String key = type.getName();
        if (mSubjectHashMap.get(key) != null) {
            mSubjectHashMap.get(key).add(subscription);
        } else {
            CompositeDisposable compositeSubscription = new CompositeDisposable();
            compositeSubscription.add(subscription);
            mSubjectHashMap.put(key, compositeSubscription);
        }
    }

    /**
     * 取消订阅
     */
    public <T> void unSubscribe(Class<T> type) {
        if (mSubjectHashMap == null) {
            return;
        }
        String key = type.getName();
        if (!mSubjectHashMap.containsKey(key)) {
            return;
        }
        if (mSubjectHashMap.get(key) != null) {
            mSubjectHashMap.get(key).dispose();
        }
        mSubjectHashMap.remove(key);
    }

    private static class SingletonHolder {
        private static RxBus instance = new RxBus();
    }
}